package gupao.concurrency.threadpool.futures;

import gupao.concurrency.utils.ThreadPoolService;
import gupao.concurrency.utils.Utils;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class CompletableFutureTicketSample {

    static ThreadPoolExecutor queryThreadPool = ThreadPoolService.getOrCreateThreadPoolExecutor("query", 32, 32);
    static ThreadPoolExecutor buyThreadPool = ThreadPoolService.getOrCreateThreadPoolExecutor("buy", 64, 64);
    static ThreadPoolExecutor msgThreadPool = ThreadPoolService.getOrCreateThreadPoolExecutor("msg", 64, 64);


    public static void main(String[] args) {
        TicketService ticketService = new TicketService("0001", 20);
        int num = 1000;
        testTicketService(ticketService, num);
    }

    static void testTicketService(TicketService ticketService, int num) {
        System.out.println(ticketService.getClass().getSimpleName() + ">>");
        CompletableFuture<Boolean>[] cfArray = new CompletableFuture[num];
        AtomicInteger successCount = new AtomicInteger(0);
        AtomicInteger failCount = new AtomicInteger(0);
        log.info("{} start to submit tasks..", ticketService.getClass());
        long x = System.currentTimeMillis();
        for (int i = 0; i < num; i++) {
            cfArray[i] = CompletableFuture.supplyAsync(ticketService::queryTicket, queryThreadPool)
                    .thenApplyAsync(count -> {
                        if (count > 0) {
                            return ticketService.buyTicket();
                        }
                        return false;
                    }, buyThreadPool).thenApplyAsync(buySuccess -> {
                        if (buySuccess) {
                            boolean result = ticketService.sendMsg();
                            if (result) {
                                successCount.incrementAndGet();
                            } else {
                                failCount.incrementAndGet();
                            }
                            return result;
                        } else {
                            failCount.incrementAndGet();
                        }
                        return false;
                    }, msgThreadPool);
        }
        log.info("{} Wait all task done.", ticketService.getClass());
        CompletableFuture.allOf(cfArray).join();
        log.info("{} Success Count: {}, fail count: {}, time cost: {}ms", ticketService.getClass().getSimpleName(), successCount.get(), failCount.get(), (System.currentTimeMillis() - x));
    }

    @AllArgsConstructor
    public static class TicketModel{
        final String name;
        volatile long count;
    }

    static class TicketService {
        final TicketModel model;
        Lock lock = new ReentrantLock();

        public TicketService(String name, long count) {
            this.model = new TicketModel(name, count);
        }

        long queryTicket() {
            lock.lock();
            try {
                Utils.sleepIgnoreExp(3);
                return model.count;
            } finally {
                lock.unlock();
            }
        }

        boolean buyTicket() {
            lock.lock();
            try {
                Utils.sleepIgnoreExp(6);
                long x = model.count - 1;
                if (x >= 0) {
                    model.count = x;
                    return true;
                }
                return false;
            } finally {
                lock.unlock();
            }
        }

        boolean sendMsg() {
            Utils.sleepIgnoreExp(1);
            return true;
        }
    }
}
